// Copyright 2023 Huawei Cloud Computing Technology Co., Ltd.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include <iostream>
#include <cstring>
#include <time.h>
#include <chrono>
#include <sys/socket.h>
#include <sys/epoll.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <net/if.h>
#include <fcntl.h>
#include <unistd.h>
#include <stdio.h>
#include <errno.h>
#include <sys/ioctl.h>
#include <algorithm>

#include "CommonMaster.h"
#include "EpollModule.h"
#include "RapidJson.h"
#include "RrConfig.h"
#include "Log.h"

namespace master {

std::map<std::string, int> uuidSocketMap;

uint64_t getCurrentTime(void)
{
    struct timeval tv;
    struct timezone tz;
    uint64_t ultime;

    gettimeofday(&tv, &tz);
    ultime = (uint64_t)(NUM_SCALE_BIG_1000000(tv.tv_sec)) + (uint64_t)(tv.tv_usec);
    return ultime;
}

uint32_t getIdleCount(void)
{
    return CommonMaster::GetInstance().getIdleWorkerCount();
}

CommonMaster::CommonMaster()
{
    pSocketServer = std::make_shared<SocketServer>();
    threadRunning = IS_RUNNING;
}

CommonMaster::~CommonMaster()
{
    LOG_INFO_FMT("~CommonMaster()");
}

int CommonMaster::findUuidSocketFdInfoIt(std::string workerUuid)
{
    std::lock_guard<std::mutex> locker(mutexUuidMap);
    int fd = 0;
    auto uuidFdMapIt = uuidSocketMap.find(workerUuid);
    if (uuidFdMapIt != uuidSocketMap.end()) {
        fd = uuidFdMapIt->second;
    } else {
        LOG_DEBUG_FMT("UuidFdMapIt is not exited");
    }
    return fd;
}

void CommonMaster::insertUuidSocketFdInfoIt(std::string workerUuid, int fd)
{
    std::lock_guard<std::mutex> locker(mutexUuidMap);
    uuidSocketMap.insert(std::pair<std::string, int>(workerUuid, fd));
    LOG_INFO_FMT("Insert workerUuid %s fd %d", workerUuid.c_str(), fd);
}

void CommonMaster::eraseUuidSocketFdInfoIt(std::string workerUuid)
{
    std::lock_guard<std::mutex> locker(mutexUuidMap);
    uuidSocketMap.erase(workerUuid);
}

int CommonMaster::ServiceThreadFunc(void)
{
    LOG_INFO_FMT("Service thread run");
    EpollModules::GetInstance().Run();
    return ENC_OK;
}

int CommonMaster::eraseHeartbeatTimeOutUuid()
{
    std::lock_guard<std::mutex> locker(workerInfoMapMutex);

    std::vector<std::string> workerUuids;
    std::string workUuid;

    uint64_t currentTime = getCurrentTime();
    uint32_t workerStatus;
    uint32_t idleCount = 0;
    uint32_t assignCount = 0;
    uint32_t useCount = 0;
    for (auto workerInfoMapIt = workerInfoMap.begin(); workerInfoMapIt != workerInfoMap.end(); workerInfoMapIt++) {
        workerStatus = workerInfoMapIt->second.status;
        statisticWorkerStatus(workerStatus, &idleCount, &assignCount, &useCount);
        if (currentTime < workerInfoMapIt->second.heartbeatTime + rr::gMasterConfig.iMaster.heartbeatTimeout) {
            continue;
        }
        workUuid = workerInfoMapIt->first;
        masterServiceClose(findUuidSocketFdInfoIt(workUuid));
        eraseUuidSocketFdInfoIt(workUuid);
        workerUuids.push_back(workUuid);
    }

    LOG_INFO_FMT("Stats worker idleCount %d assignCount %d useCount %d", idleCount, assignCount, useCount);
    for (auto workerUuidsIt = workerUuids.begin(); workerUuidsIt != workerUuids.end(); workerUuidsIt++) {
        workUuid = *workerUuidsIt;
        workerInfoMap.erase(workUuid);
    }

    return ENC_OK;
}

void CommonMaster::statisticWorkerStatus(uint32_t workerStatus, uint32_t *idleCount, uint32_t *assignCount, uint32_t *useCount)
{
    switch (workerStatus) {
        case 0:
            (*idleCount)++;
            break;
        case 1:
            (*assignCount)++;
            break;
        case 2:
            (*useCount)++;
            break;
        default:
            LOG_INFO_FMT("Invalid status %d", workerStatus);
            break;
    }
}

void CommonMaster::eraseNotExistUuid()
{
    std::lock_guard<std::mutex> locker(workerInfoMapMutex);

    std::string workUuid;
    for (auto uuidFdMapIt = uuidSocketMap.begin(); uuidFdMapIt != uuidSocketMap.end(); uuidFdMapIt++) {
        workUuid = uuidFdMapIt->first;
        auto it = workerInfoMap.find(workUuid);
        if (it == workerInfoMap.end()) {
            this->masterServiceClose(findUuidSocketFdInfoIt(workUuid));
            this->eraseUuidSocketFdInfoIt(workUuid);
        }
    }
}

uint32_t CommonMaster::getIdleWorkerCount()
{
    std::lock_guard<std::mutex> locker(workerInfoMapMutex);
    uint32_t idleWorkerCount = 0;
    for (auto workerInfoMapIt = workerInfoMap.begin(); workerInfoMapIt != workerInfoMap.end(); workerInfoMapIt++) {
        if (workerInfoMapIt->second.status == 0) {
            idleWorkerCount++;
        }
    }
    return idleWorkerCount;
}

int CommonMaster::getPrivateIp(const std::string &netName, std::string &privateIp)
{
    int sock_fd;
    struct ifconf conf;
    struct ifreq *ifr;
    char buff[DEFAULT_BUF_SIZE] = {0};
    int num;
    int i;
    sock_fd = socket(PF_INET, SOCK_DGRAM, 0);
    if (sock_fd < 0) {
        LOG_ERROR_FMT("Failed to create socket, errno=%d", errno);
        return -1;
    }
    conf.ifc_len = DEFAULT_BUF_SIZE;
    conf.ifc_buf = buff;
    if (ioctl(sock_fd, SIOCGIFCONF, &conf) < 0) {
        LOG_ERROR_FMT("Failed to set ifconf, errno=%d", errno);
        close(sock_fd);
        return -1;
    }

    num = conf.ifc_len / sizeof(struct ifreq);
    ifr = conf.ifc_req;
    for (i = 0; i < num; i++) {
        struct sockaddr_in *sin = (struct sockaddr_in *)(&ifr->ifr_addr);
        if (ioctl(sock_fd, SIOCGIFFLAGS, ifr) < 0) {
            LOG_ERROR_FMT("Failed to get ifflags, errno=%d", errno);
            close(sock_fd);
            return -1;
        }
        if ((ifr->ifr_flags & IFF_UP) && strcmp(netName.c_str(),ifr->ifr_name) == 0) {
            privateIp = inet_ntoa(sin->sin_addr);
            close(sock_fd);
            return 0;
        }
        ifr++;
    }
    close(sock_fd);
    return -1;
}

void CommonMaster::ServiceWorkerHeartBeatFunc(void)
{
    shared_ptr<WorkContextMgT> pWorkerCapInfo = make_shared<WorkContextMgT>();
    if (pWorkerCapInfo == nullptr) {
        return;
    }

    while (IS_RUNNING == threadRunning) {
        RapidJson pJson;
        std::string sWorkerCapRsp;
        std::vector<std::string> vecWorkerUuid;

        this->eraseHeartbeatTimeOutUuid();
        this->eraseNotExistUuid();

        sleep(DEFAULT_HEARTBEAT_CHECK_INTERNAL);
    }
}

void *CommonMaster::ServiceThreadInit(void *pArgc)
{
    CommonMaster *pThis = (CommonMaster *)pArgc;
    pThis->ServiceThreadFunc();
    return pThis;
}

void *CommonMaster::ServiceHeartBeatThreadInit(void *pArgc)
{
    CommonMaster *pThis = (CommonMaster *)pArgc;
    pThis->ServiceWorkerHeartBeatFunc();
    return pThis;
}

int CommonMaster::createThreadInit(void)
{
    if (ENC_OK != pthread_create(&pthreadId, NULL, ServiceThreadInit, this)) {
        LOG_ERROR_FMT("pthread_create pthreadId = %d fail", pthreadId);
        return ENC_FAIL;
    }

    if (ENC_OK != pthread_create(&heartPthreadId, NULL, ServiceHeartBeatThreadInit, this)) {
        LOG_ERROR_FMT("pthread_create heartPthreadId %d fail", heartPthreadId);
        return ENC_FAIL;
    }

    return ENC_OK;
}

int CommonMaster::masterServiceThreadMutexInit()
{
    if (ENC_OK != pthread_mutex_init(&mutex, NULL)) {
        LOG_ERROR_FMT("pthread_mutex_init failed");
        pthread_mutex_destroy(&mutex);
        return ENC_FAIL;
    }

    return ENC_OK;
}

int CommonMaster::handleWorkerHeartbeat(std::shared_ptr<RedisOperator> &redisClient, uint8_t *pOutData, uint32_t outLen, int fd, std::string requestId)
{
    std::lock_guard<std::mutex> locker(workerInfoMapMutex);

    int ret = ENC_OK;
    WorkerStatusRspT pWorkerCap;
    ret = memcpy_s(&pWorkerCap, WORKER_CAPACITY_RSP_SIZE, pOutData, outLen);
    if (ret < 0) {
        LOG_ERROR_FMT("Parse heartbeat mem copy failed");
        return ENC_FAIL;
    }

    std::string keyUuid = pWorkerCap.workerUuid;
    uint64_t heartbeatTime = getCurrentTime();
    auto it = workerInfoMap.find(keyUuid);
    if (it == workerInfoMap.end()) {
        WorkContextMgT workerInfo;
        workerInfo.ip = pWorkerCap.workerIp;
        workerInfo.port = pWorkerCap.port;
        workerInfo.status = pWorkerCap.status;
        workerInfo.encodeType = pWorkerCap.encodeType;
        workerInfo.resourceType = pWorkerCap.resourceType;
        workerInfo.dataType = pWorkerCap.dataType;
        workerInfo.flavorType = pWorkerCap.flavorType;
        workerInfo.frameRate = pWorkerCap.frameRate;
        workerInfo.fd = fd;
        workerInfo.heartbeatTime = heartbeatTime;
        workerInfo.assignTime = 0;
        workerInfo.phoneId = "";
        workerInfoMap.insert(std::pair<std::string, WorkContextMgT>(keyUuid, workerInfo));
        return ENC_OK;
    }
    workerInfoMap[keyUuid].heartbeatTime = heartbeatTime;

    uint64_t currentTime = getCurrentTime();
    int timeout = (int) currentTime - (int) workerInfoMap[keyUuid].assignTime;
    if (workerInfoMap[keyUuid].status == 1 && timeout >= DEFAULT_ASSIGN_TIME) {
        LOG_WARN_FMT("Worker %s assign time out %ld", keyUuid.c_str(), timeout);
        workerInfoMap[keyUuid].status = 0;
    }

    return ENC_OK;
}

int CommonMaster::cloudPhoneSchedulingRsp(shared_ptr <WorkContextMgT> &pWorkerContext, RemoteSchedulingRspT *pPhoneRsp, int fd, std::string requestId)
{
    errno_t err = strcpy_s((char *)pPhoneRsp->ipStr, pWorkerContext->ip.length() + 1, pWorkerContext->ip.c_str());
    if (err != ENC_OK) {
        LOG_ERROR_FMT("Server req cloud phone str cpy %s failed", pPhoneRsp->ipStr);
        return ENC_FAIL;
    }

    pPhoneRsp->port = pWorkerContext->port;
    pPhoneRsp->result = SUCCESS;
    pPhoneRsp->encodeType = pWorkerContext->encodeType;

    LOG_INFO_FMT("ReqId %s req end assign", requestId.c_str());

    EpollModules::GetInstance().StoreSendData(reinterpret_cast<uint8_t *>(pPhoneRsp), 
            sizeof(RemoteSchedulingRspT), fd, CLOUD_PHONE_DEVICE_RSP, requestId);
    
    return ENC_OK;
}

int CommonMaster::cloudPhoneSchedulingReq(std::shared_ptr<RedisOperator> &redisClient, uint8_t *pOutData, uint32_t outLen, int fd, std::string requestId)
{
    int ret = ENC_OK;
    RemoteSchedulingReqT gPhoneReq;
    RemoteSchedulingRspT gPhoneRsp;

    ret = memcpy_s(&gPhoneReq, sizeof(RemoteSchedulingReqT), pOutData, outLen);
    if (ret < 0) {
        LOG_ERROR_FMT("Server req cloud phone mem copy failed");
        return ENC_FAIL;
    }

    uint8_t protocolVersion = outLen == INIT_REMOTE_SCHEDULING_REQ_SIZE ? 0 : gPhoneReq.protocolVersion;

    std::string cmdContent = getCmdContent(gPhoneReq.resourceType, gPhoneReq.dataType, gPhoneReq.flavorType, gPhoneReq.frameRate);
    LOG_INFO_FMT("ReqId %s req phone %s phoneId=%s version=%d", requestId.c_str(), cmdContent.c_str(), gPhoneReq.phoneId, protocolVersion);

    auto pSocketInfo = pSocketServer->findSocketInfoIt(fd);
    if (pSocketInfo == nullptr) {
        LOG_ERROR_FMT("Server req cloud phone find socket info failed");
        return ENC_FAIL;
    }

    auto pWorkerContext = make_shared<WorkContextMgT>();
    if (pWorkerContext == nullptr) {
        LOG_ERROR_FMT("Server req cloud phone context is null");
        return ENC_FAIL;
    }

    ret = this->selectWorker(redisClient, &gPhoneReq, pWorkerContext, protocolVersion, requestId);
    if (SUCCESS != ret) {
        gPhoneRsp.result = ret;
        strcpy(gPhoneRsp.ipStr, REMOTE_SCHEDULING_DEFAULT_IP.c_str());
        gPhoneRsp.port = REMOTE_SCHEDULING_DEFAULT_PORT;
        gPhoneRsp.encodeType = CPU;
        EpollModules::GetInstance().StoreSendData(reinterpret_cast<uint8_t *>(&gPhoneRsp),
            sizeof(RemoteSchedulingRspT), fd, CLOUD_PHONE_DEVICE_RSP, requestId);

        return ENC_OK;
    }

    if (ENC_OK != this->cloudPhoneSchedulingRsp(pWorkerContext, &gPhoneRsp,fd, requestId)) {
        LOG_ERROR_FMT("Server cloud phone rsp failed");
        return ENC_FAIL;
    }

    LOG_INFO_FMT("ReqId %s req phone end", requestId.c_str());
    return ENC_OK;
}

int CommonMaster::firstReportWorkerStatus(WorkerStatusRspT &pWorkerCap, shared_ptr<WorkContextMgT> pWorkerDevice,
                                            int fd, std::string requestId)
{
    WorkerStatusReqT gWorkerCapacityReq;
    errno_t err = strcpy_s(gWorkerCapacityReq.workerUuid, UUID_LENGTH, pWorkerCap.workerUuid);
    if (err != ENC_OK) {
        LOG_ERROR_FMT("strcpy %s failed!", gWorkerCapacityReq.workerUuid);
        return ENC_FAIL;
    }

    this->insertUuidSocketFdInfoIt(string(gWorkerCapacityReq.workerUuid), fd);
    return ENC_OK;
}

int CommonMaster::reportWorkerStatusInsertMem(std::shared_ptr<RedisOperator> &redisClient, WorkerStatusRspT &pWorkerCap,shared_ptr<WorkContextMgT> pWorkerDevice,
                                                  int fd)
{
    std::lock_guard<std::mutex> locker(workerInfoMapMutex);

    std::string keyUuid = pWorkerCap.workerUuid;
    uint64_t heartbeatTime = getCurrentTime();

    auto it = workerInfoMap.find(keyUuid);
    if (it != workerInfoMap.end()) {
        workerInfoMap[keyUuid].status = pWorkerCap.status;
        workerInfoMap[keyUuid].fd = fd;
        workerInfoMap[keyUuid].heartbeatTime = heartbeatTime;
        if (pWorkerDevice->status != 2) {
            workerInfoMap[keyUuid].phoneId = "";
        }
    } else {
        WorkContextMgT workerInfo;
        workerInfo.ip = pWorkerCap.workerIp;
        workerInfo.port = pWorkerCap.port;
        workerInfo.status = pWorkerCap.status;
        workerInfo.encodeType = pWorkerCap.encodeType;
        workerInfo.resourceType = pWorkerCap.resourceType;
        workerInfo.dataType = pWorkerCap.dataType;
        workerInfo.flavorType = pWorkerCap.flavorType;
        workerInfo.frameRate = pWorkerCap.frameRate;
        workerInfo.fd = fd;
        workerInfo.heartbeatTime = heartbeatTime;
        workerInfo.assignTime = 0;
        workerInfo.phoneId = "";
        workerInfoMap.insert(std::pair<std::string, WorkContextMgT>(keyUuid, workerInfo));
    }

    LOG_INFO_FMT(
            "Worker update status %s, status=%d, heartbeatTime=%ld",
            getWorkUuidContent(keyUuid).c_str(), pWorkerCap.status, heartbeatTime);
    return ENC_OK;
}

int CommonMaster::handleWorkerUpdateStatus(std::shared_ptr<RedisOperator> &redisClient, uint8_t *pOutData, uint32_t outLen, int fd, std::string requestId)
{
    int ret = ENC_OK;
    WorkerStatusRspT pWorkerCap;
    shared_ptr<WorkContextMgT> pWorkerDevice = make_shared<WorkContextMgT>();
    if (pWorkerDevice == nullptr) {
        LOG_ERROR_FMT("Capacity inquire device is null");
        return ENC_FAIL;
    }

    ret = memcpy_s(&pWorkerCap, WORKER_CAPACITY_RSP_SIZE, pOutData, outLen);
    if (ret < 0) {
        LOG_ERROR_FMT("Capacity inquire mem copy failed");
        return ENC_FAIL;
    }

    auto pSocketInfo = pSocketServer->findSocketInfoIt(fd);
    if (nullptr == pSocketInfo) {
        LOG_ERROR_FMT("Capacity inquire find socket infoId failed");
        return ENC_FAIL;
    }

    if(pSocketInfo->firstRun){
        pSocketInfo->firstRun = false;
        if(this->firstReportWorkerStatus(pWorkerCap, pWorkerDevice, fd, requestId) != ENC_OK){
            LOG_ERROR_FMT("First inquire cap failed");
            return ENC_OK;
        }
    }

    if(this->reportWorkerStatusInsertMem(redisClient, pWorkerCap, pWorkerDevice, fd) != ENC_OK){
        LOG_ERROR_FMT("Inquire cap insert redis failed");
        return ENC_FAIL;
    }

    return ENC_OK;
}

int CommonMaster::serverParseData(std::shared_ptr<RedisOperator> &redisClient, uint16_t cmdCtl, uint8_t *pOutData, uint32_t outLen, int fd,std::string requestId)
{
    switch (cmdCtl) {
        case WORKER_REPORT_HEARTBEAT:
            this->handleWorkerHeartbeat(redisClient, pOutData, outLen, fd, requestId);
            break;
        case CLOUD_PHONE_DEVICE_REQ:
            if (ENC_OK != this->cloudPhoneSchedulingReq(redisClient, pOutData, outLen, fd, requestId)) {
                LOG_ERROR_FMT("Server req cloud phone device failed");
                return ENC_FAIL;
            }
            break;
        case WORKER_REPORT_STATUS_RSP:
            if (ENC_OK != this->handleWorkerUpdateStatus(redisClient, pOutData, outLen, fd, requestId)) {
                LOG_ERROR_FMT("Server inquire work capacity failed");
                return ENC_FAIL;
            }
            break;
        default:
            break;
    }

    return ENC_OK;
}

int CommonMaster::selectWorker(std::shared_ptr<RedisOperator> &redisClient, RemoteSchedulingReqT *pPhoneReq,
                                              shared_ptr<WorkContextMgT> &pWorkerCapInfo, uint8_t protocolVersion, std::string requestId)
{
    std::lock_guard<std::mutex> locker(workerInfoMapMutex);

    std::string workerUuid;
    std::vector<std::string> workerUuids;
    for (auto workerInfoMapIt = workerInfoMap.begin(); workerInfoMapIt != workerInfoMap.end(); workerInfoMapIt++) {
        if (workerInfoMapIt->second.status != 0) {
            continue;
        }
        workerUuid = workerInfoMapIt->first;
        workerUuids.push_back(workerUuid);
    }

    uint32_t validUuidCount = workerUuids.size();
    if (validUuidCount <= 0) {
        LOG_ERROR_FMT("ReqId %s master select no available worker", requestId.c_str());
        return NO_AVAILABLE;
    }
    uint32_t randomIndex = rand() % validUuidCount;
    workerUuid = workerUuids.at(randomIndex);

    if (!isSupportSchedulingReq(workerInfoMap[workerUuid], pPhoneReq, protocolVersion, requestId)) {
        LOG_ERROR_FMT("ReqId %s master select no match worker", requestId.c_str());
        return NO_MATCH;
    }

    workerInfoMap[workerUuid].status = 1;
    workerInfoMap[workerUuid].phoneId = pPhoneReq->phoneId;
    workerInfoMap[workerUuid].assignTime = getCurrentTime();

    pWorkerCapInfo->ip = workerInfoMap[workerUuid].ip;
    pWorkerCapInfo->port = workerInfoMap[workerUuid].port;
    pWorkerCapInfo->encodeType = workerInfoMap[workerUuid].encodeType;

    LOG_INFO_FMT("ReqId %s master select worker %s", requestId.c_str(), getWorkUuidContent(workerUuid).c_str());
    return SUCCESS;
}

bool CommonMaster::isSupportSchedulingReq(WorkContextMgT workerContext, RemoteSchedulingReqT *pPhoneReq, uint8_t protocolVersion, std::string requestId)
{
    int supportResourceType = workerContext.resourceType;
    int supportDataType = workerContext.dataType;
    int supportFlavorType = workerContext.flavorType;
    int supportFrameRate = workerContext.frameRate;

    int reqResourceType = pPhoneReq->resourceType;
    int reqDataType = pPhoneReq->dataType;
    int reqFlavorType = pPhoneReq->flavorType;
    int reqFrameRate = pPhoneReq->frameRate;

    if (supportResourceType != reqResourceType) {
        return false;
    }

    if (supportDataType == H264 && reqDataType > H264) {
        LOG_ERROR_FMT("ReqId %s encode type do not support", requestId.c_str());
        return false;
    }

    if (supportFlavorType < reqFlavorType) {
        LOG_ERROR_FMT("ReqId %s resolution type do not support", requestId.c_str());
        return false;
    }

    if (protocolVersion != 0 && (supportFrameRate < reqFrameRate)) {
        LOG_ERROR_FMT("ReqId %s frame rate do not support", requestId.c_str());
        return false;
    }

    return true;
}

std::string CommonMaster::getWorkUuidContent(std::string workUuid)
{
    std::string uuid = workUuid;
    std::stringstream uuidStream(uuid);
    std::vector<std::string> sections;

    char delimiter = ':';
    std::string section;
    while (std::getline(uuidStream, section, delimiter)) {
        sections.push_back(section);
    }

    string workerIp = sections.at(0);
    string workerPort = sections.at(1);
    int encodeType = std::stoi(sections.at(2));
    int resourceType = std::stoi(sections.at(3));
    int dataType = std::stoi(sections.at(4));
    int flavorType = std::stoi(sections.at(5));
    int frameRate = std::stoi(sections.at(6));

    string worker = workerIp + DEFAULT_DELIMITER + workerPort + DEFAULT_DELIMITER + (encodeType == CPU ? "CPU" : "GPU");
    string cmdContent = getCmdContent(resourceType, dataType, flavorType, frameRate);
    return worker + DEFAULT_SPACE + cmdContent;
}

std::string CommonMaster::getCmdContent(int resourceType, int dataType, int flavorType, int frameRate)
{
    string resource;
    string data;
    string flavor;
    string fps;

    switch(resourceType) {
        case ENCODE:
            resource = "ENCODE";
            break;
        case DECODE:
            resource = "DECODE";
            break;
        default :
            resource = "INVALID";
            LOG_ERROR_FMT("Invalid resource type");
            break;
    }

    switch(dataType) {
        case H264:
            data = "H264";
            break;
        case H265:
            data = "H265";
            break;
        case MIX:
            data = "MIX";
            break;
        default :
            data = "INVALID";
            LOG_ERROR_FMT("Invalid data type");
            break;
    }

    switch(flavorType) {
        case P720:
            flavor = "720P";
            break;
        case P1080:
            flavor = "1080P";
            break;
        default :
            flavor = "INVALID";
            LOG_ERROR_FMT("Invalid flavor type");
            break;
    }

    fps = to_string(frameRate);

    string cmd = "cmd=" + to_string(resourceType) + DEFAULT_DELIMITER + to_string(dataType)
            + DEFAULT_DELIMITER + to_string(flavorType) + DEFAULT_DELIMITER + fps;
    string content = "content=" + resource + DEFAULT_DELIMITER + data + DEFAULT_DELIMITER + flavor + DEFAULT_DELIMITER + fps;
    return cmd + DEFAULT_SPACE + content;
}

int CommonMaster::commonMasterReadDataAnalysis(int fd, std::string requestId)
{
    int ret = ENC_OK;
    int cmdCtl = 0;

    auto pSocketInfo = pSocketServer->findSocketInfoIt(fd);
    if (nullptr == pSocketInfo) {
        return ENC_FAIL;
    }

    ret = pSocketServer->socketServerReadData(pSocketInfo);
    if (ENC_OK != ret) {
        this->masterServiceClose(pSocketInfo->fdSock);
        return ENC_FAIL;
    }

    if (ENC_OK == memcmp(pSocketInfo->recvBuf.get(), &SOCKET_HEAD_FLAG, SOCKET_HEAD_FLAG_LENGTH)) {
        ret = memcpy_s(&cmdCtl, SOCKET_CMD_LENGTH, pSocketInfo->recvBuf.get() + SOCKET_HEAD_FLAG_LENGTH,
                       SOCKET_CMD_LENGTH);
        if (ret < 0) {
            LOG_ERROR_FMT("Read data analyze memory copy failed");
            return ENC_FAIL;
        }

        if (cmdCtl == CLOUD_PHONE_DEVICE_REQ) {
            LOG_INFO_FMT("ReqId %s req wait assign", requestId.c_str());
        }

        EpollModules::GetInstance().pQueue->Push(QUEUE_DEAL, pSocketInfo->recvBuf.get() + SOCKET_HEAD_LENGTH,
                                                 pSocketInfo->recvBufLen - SOCKET_HEAD_LENGTH, cmdCtl, pSocketInfo->fdSock, requestId);
    }
    return ENC_OK;
}

int CommonMaster::commonMasterWriteData(uint16_t cmdCtl, uint8_t *inBuf, uint32_t inLen, int fd, std::string requestId)
{
    int ret = ENC_OK;
    uint32_t protocolLen = 0;

    if ((inBuf == nullptr) || (inLen <= 0) || (inLen > DST_BUF_LEN - SOCKET_BUF_BLOCK)) {
        LOG_ERROR_FMT("Write data param invalid");
        return ENC_FAIL;
    }

    auto pSocketInfo = pSocketServer->findSocketInfoIt(fd);
    if (pSocketInfo == nullptr) {
        LOG_ERROR_FMT("Write data find fd is null");
        return ENC_FAIL;
    }

    ret = memcpy_s(pSocketInfo->sendBuf.get(), SOCKET_HEAD_FLAG_LENGTH + 1, (uint8_t *) &SOCKET_HEAD_FLAG,
                   SOCKET_HEAD_FLAG_LENGTH);
    if (ret < 0) {
        LOG_ERROR_FMT("Write data mem copy failed");
        return ENC_FAIL;
    }

    ret = memcpy_s(pSocketInfo->sendBuf.get() + SOCKET_HEAD_FLAG_LENGTH, SOCKET_CMD_LENGTH + 1, &cmdCtl,
                   SOCKET_CMD_LENGTH);
    if (ret < 0) {
        LOG_ERROR_FMT("Write data mem copy failed");
        return ENC_FAIL;
    }

    protocolLen = inLen;
    ret = memcpy_s(pSocketInfo->sendBuf.get() + SOCKET_HEAD_FLAG_LENGTH + SOCKET_CMD_LENGTH, SOCKET_DATA_BUF_LENGTH + 1, &protocolLen, SOCKET_DATA_BUF_LENGTH);
    if (ret < 0) {
        LOG_ERROR_FMT("Write data mem copy failed");
        return ENC_FAIL;
    }

    ret = memcpy_s(pSocketInfo->sendBuf.get() + SOCKET_HEAD_LENGTH, inLen, inBuf, inLen);
    if (ret < 0) {
        LOG_ERROR_FMT("Write data mem copy failed");
        return ENC_FAIL;
    }
    pSocketInfo->sendBufLen = SOCKET_HEAD_LENGTH + inLen;

    ret = pSocketServer->socketServerWriteData(fd, pSocketInfo);
    if (ENC_OK != ret) {
        LOG_ERROR_FMT("Write data failed");
        this->masterServiceClose(fd);
        return ENC_FAIL;
    }
    return ENC_OK;
}

int CommonMaster::masterServiceInit(uint16_t ipPort)
{
    if (ENC_OK != pSocketServer->socketServerInit(ipPort)) {
        LOG_ERROR_FMT("Server socket init failed");
        pSocketServer->socketClose();
        return ENC_FAIL;
    }
    LOG_INFO_FMT("Server socket init success");

    if (ENC_OK != this->createThreadInit()) {
        LOG_ERROR_FMT("Create thread init failed");
        return ENC_FAIL;
    }

    this->masterServiceThreadCondInit();
    if (ENC_OK != this->masterServiceThreadMutexInit()) {
        LOG_ERROR_FMT("Master thread mutex init fail");
        return ENC_FAIL;
    }

    LOG_INFO_FMT("Master service init success");
    return ENC_OK;
}

int CommonMaster::dealConfig(const std::string configFilePath)
{
    rr::RrConfig config;

    if (!config.ReadConfig(configFilePath)) {
        LOG_ERROR_FMT("Read config file failed, use the default configuration");
        return ENC_FAIL;
    }

    getPrivateIp("eth0", rr::gMasterConfig.iMaster.privateIp);
    LOG_INFO_FMT("Private ip is %s", rr::gMasterConfig.iMaster.privateIp.c_str());

    rr::gMasterConfig.iMaster.port =
            config.ReadInt("MASTER", "port", rr::DEFAULT_MASTER_PORT);
    rr::gMasterConfig.iMaster.healthCheckPort =
            config.ReadInt("MASTER", "healthCheckPort", rr::DEFAULT_HEALTH_CHECK_PORT);
    rr::gMasterConfig.iMaster.healthCheckUrl =
            config.ReadString("MASTER", "healthCheckUrl", rr::DEFAULT_HEALTH_CHECK_URL.c_str());
    rr::gMasterConfig.iMaster.heartbeatTimeout =
            config.ReadInt("MASTER", "heartbeatTimeout", rr::DEFAULT_HEARTBEAT_TIMEOUT);

    LOG_INFO_FMT("Deal config file success port=%d healthCheckPort=%d healthCheckUrl=%s heartbeatTimeout=%d", rr::gMasterConfig.iMaster.port, rr::gMasterConfig.iMaster.healthCheckPort,
                  rr::gMasterConfig.iMaster.healthCheckUrl.c_str(), rr::gMasterConfig.iMaster.heartbeatTimeout);
    return ENC_OK;
}

int CommonMaster::masterServiceClose(int fd)
{
    EpollModules::GetInstance().DeleteFd(fd, EV_READ);
    if (ENC_OK != pSocketServer->eraseSocketInfoIt(fd)) {
        return ENC_FAIL;
    }
    return ENC_OK;
}

int masterServiceStartInit(std::string configFileName)
{
    if (ENC_OK != CommonMaster::GetInstance().dealConfig(configFileName)) {
        LOG_ERROR_FMT("Read config file failed");
        return ENC_FAIL;
    }

    if (ENC_OK != CommonMaster::GetInstance().masterServiceInit(rr::gMasterConfig.iMaster.port)) {
        LOG_ERROR_FMT("Master service init failed");
        return ENC_FAIL;
    }

    EpollModules::GetInstance().Init(EPOLL_MAX_EVENT);
    return ENC_OK;
}
};